#include "config.h"

#include <ctype.h>
#include <errno.h>

#define DBG_SUBSYS S_LIBYNET

#include "adt.h"
#include "ynet_net.h"
#include "ynet_rpc.h"
#include "../sock/sock_udp.h"
#include "net_msg.h"
#include "dbg.h"

int net_sendmulti(const int *sd, int count, const char *req, int reqlen)
{
        int ret, i, len;
        char buf[BUF_SIZE_8K];
        msg_head_t *head;

        YASSERT(reqlen <= MAX_BUF_LEN);

        head = (void *)buf;
        head->len = reqlen;
        head->crc = crc32_sum(req, reqlen);
        memcpy(head->buf, req, reqlen);

        len = reqlen + sizeof(msg_head_t);

        for (i = 0; i < count; i++) {
                ret = _send(sd[i], buf, len, MSG_NOSIGNAL);
                if (ret < 0) {
                        DWARN("send fail\n");
                        continue;
                }
        }

        return 0;
}

static int __net_recv(const struct pollfd *pfd, msg_head_t *head)
{
        int ret, retry = 0;

retry:
        ret = _recv(pfd->fd, head, BUF_SIZE_8K, MSG_PEEK);
        if (ret < 0) {
                ret = -ret;
                YASSERT(ret != EAGAIN);
                GOTO(err_ret, ret);
        }

        if ((unsigned)ret != sizeof(msg_head_t) + head->len) {
                YASSERT(retry < 100);
                DWARN("msg got %u need %llu\n", ret,
                      (LLU)sizeof(msg_head_t) + head->len);
                usleep(100);
                goto retry;
        }

        ret = head->retval;
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int net_recvmulti(const int *sd, int count, char *rep, int *replen, int usec, uuid_t *uuid)
{
        int ret, pcount, i, done = 0;
        struct pollfd pfds[PAXOS_PEER_MAX], *pfd;
        msg_head_t *head;

        char *buf = mem_cache_calloc1(MEM_CACHE_8K, PAGE_SIZE);

        YASSERT(count <= PAXOS_PEER_MAX);

        pcount = count;
        ret = sock_poll_sd1(sd, count, POLLIN, usec, pfds, &pcount);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        YASSERT(pcount);
        head = (void *)buf;
        for (i = 0; i < count; i++) {
                pfd = &pfds[i];

                if (pfd->revents == POLLIN) {
                        ret = __net_recv(pfd, head);
                        if (unlikely(ret))
                                GOTO(err_ret, ret);

                        done = 1;
                        break;
                }
        }

        if (done == 0) {
                ret = ENONET;
                goto err_ret;
        }

        if (head->crc != crc32_sum(head->buf, head->len)) {
                DWARN("crc not eq\n");
                ret = EAGAIN;
                GOTO(err_ret, ret);
        }

        if (uuid) {
                if (uuid_compare(head->uuid, *uuid)) {
                        DWARN("uuid not eq\n");
                        ret = EAGAIN;
                        GOTO(err_ret, ret);
                }
        }

        if (rep)
                memcpy(rep, head->buf, head->len);
        if (replen)
                *replen = head->len;

        mem_cache_free(MEM_CACHE_8K, buf);

        return 0;
err_ret:
        mem_cache_free(MEM_CACHE_8K, buf);
        return ret;
}
